package com.kakarota.hadoop.hdfs;

import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.fs.FSDataOutputStream;
import org.apache.hadoop.fs.FileStatus;
import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.fs.Path;
import org.apache.hadoop.io.IOUtils;

import java.io.IOException;
import java.io.InputStream;
import java.sql.Timestamp;
import java.util.ArrayList;
import java.util.List;

public class HadoopCommand {

    public static void main(String[] args) throws IOException {
        Configuration conf = new Configuration();
        // 设置 HDFS 访问地址
        conf.set("fs.default.name", "hdfs://192.168.1.117:9000");
        // 取得 FileSystem 文件系统实例
        FileSystem fs = FileSystem.get(conf);
        List<String> filesUnderFolder = getFileList(fs, new Path("hdfs:/"));
        filesUnderFolder.forEach(System.out::println);

        String path = "hdfs:/fileAbc.txt";
        String context = "1234";
        createFile(path, context);
        fileStatusCat(path);

    }

    public static List<String> getFileList(FileSystem fs, Path folderPath) throws IOException {
        List<String> paths = new ArrayList();

        if (fs.exists(folderPath)) {
            FileStatus[] fileStatus = fs.listStatus(folderPath);

            for (int i = 0; i < fileStatus.length; i++) {
                FileStatus fileStatu = fileStatus[i];
                paths.add(fileStatu.getPath().toString());
            }
        }

        return paths;
    }

    /**
     * 获取文件或目录的元数据信息
     */
    public static void fileStatusCat(String pathString) throws IOException {
        // 创建 Configuration 对象
        Configuration conf = new Configuration();
        // 设置 HDFS 访问地址
        conf.set("fs.default.name", "hdfs://centos01:9000");
        // 取得 FileSystem 文件系统实例
        FileSystem fs = FileSystem.get(conf);
        FileStatus fileStatus = fs.getFileStatus(new Path(pathString));

        // 判断是文件夹还是文件
        if (fileStatus.isDirectory()) {
            System.out.println("这是一个文件夹");
        } else {
            System.out.println("这是一个文件");
        }

        // 输出元数据信息
        System.out.println("文件路径：" + fileStatus.getPath());
        System.out.println("文件修改日期：" + new Timestamp(fileStatus.getModificationTime()).toString());
        System.out.println("文件上次访问日期：" + new Timestamp(fileStatus.getAccessTime()).toString());
        System.out.println("文件长度：" + fileStatus.getLen());
        System.out.println("文件备份数：" + fileStatus.getReplication());
        System.out.println("文件块大小：" + fileStatus.getBlockSize());
        System.out.println("文件所有者：" + fileStatus.getOwner());
        System.out.println("文件所在分组：" + fileStatus.getGroup());
        System.out.println("文件的权限：" + fileStatus.getPermission().toString());
    }

    /**
     * 定义创建文件方法
     */
    public static void createFile(String pathString, String context) throws IOException {
        Configuration conf = new Configuration();
        conf.set("fs.default.name", "hdfs://centos01:9000");
        FileSystem fs = FileSystem.get(conf);
        // 打开一个输出流
        FSDataOutputStream outputStream = fs.create(new Path(pathString));
        // 写入文件内容
        outputStream.write(context.getBytes());
        outputStream.close();
        fs.close();
        System.out.println("文件创建成功");
    }

    /**
     * 查询 HDFS 文件内容并输出
     */
    public static void fileSystemCat(String pathString) throws IOException {
        Configuration conf = new Configuration();
        // 设置 HDFS 访问地址
        conf.set("fs.default.name", "hdfs://centos01:9000");
        // 取得 FileSystem 文件系统实例
        FileSystem fs = FileSystem.get(conf);
        // 打开文件输入流
        InputStream in = fs.open(new Path(pathString));
        // 输出文件内容
        IOUtils.copyBytes(in, System.out, 4096, false);
        // 关闭输入流
        IOUtils.closeStream(in);
    }

    /**
     * 删除文件
     */
    public static void deleteFile(String pathString) throws IOException {
        Configuration conf = new Configuration();
        conf.set("fs.default.name", "hdfs://centos01:9000");
        FileSystem fs = FileSystem.get(conf);
        Path path = new Path(pathString);
        // 删除文件
        boolean isok = fs.deleteOnExit(path);

        if (isok) {
            System.out.println("删除成功");
        } else {
            System.out.println("删除失败");
        }

        fs.close();
    }

    /**
     * 上传本地文件到 HDFS
     */
    public static void uploadFileToHDFS(String srcPath, String dstPath) throws IOException {
        // 创建配置器
        Configuration conf = new Configuration();
        conf.set("fs.default.name", "hdfs://centos01:9000");
        // 取得 FileSystem 文件系统实例
        FileSystem fs = FileSystem.get(conf);
        // 创建可供 hadoop 使用的文件系统路径
        // 本地目录/文件
        Path src = new Path(srcPath);
        // HDFS 目录/文件
        Path dst = new Path(dstPath);
        // 复制上传文帝文件至 HDFS 文件系统中
        fs.copyFromLocalFile(src, dst);
        System.out.println("文件上传成功");
    }

    /**
     * 下载文件到本地
     */
    public static void downloadFileToLocal(String srcPath, String dstPath) throws IOException {
        // 创建配置器
        Configuration conf = new Configuration();
        conf.set("fs.default.name", "hdfs://centos01:9000");
        // 取得 FileSystem 文件系统实例
        FileSystem fs = FileSystem.get(conf);
        // 创建可供 hadoop 使用的文件系统路径
        Path src = new Path(srcPath);
        Path dst = new Path(dstPath);
        // 从 HDFS 文件系统中复制下载文件至本地
        fs.copyToLocalFile(false, src, dst, true);
        System.out.println("文件下载成功");
    }

}
